package juc.blockqueue.priorityblockingqueue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;

/**
 * @ClassName DelayedMessageProcessor
 * @PackageName juc.blockqueue.priorityblockingqueue
 * @projectName myThread
 * @Description 3. 延迟消息处理
 * @Author games
 * @Date 2025/9/4 上午11:35
 * @Version 1.0
 */
class DelayedMessage implements Comparable<DelayedMessage> {
    private final String message;
    private final long executeTime;

    public DelayedMessage(String message, long delayMillis) {
        this.message = message;
        this.executeTime = System.currentTimeMillis() + delayMillis;
    }

    @Override
    public int compareTo(DelayedMessage other) {
        //(x < y) ? -1 : ((x == y) ? 0 : 1);
        return Long.compare(this.executeTime, other.executeTime);
    }

    public boolean isReady() {
        return System.currentTimeMillis() >= executeTime;
    }

    // getter...
    public String getMessage() { return message; }
}

public class DelayedMessageProcessor {
    public static void main(String[] args) {
        BlockingQueue<DelayedMessage> messageQueue = new PriorityBlockingQueue<>();

        // 消息发送者
        new Thread(() -> {
            try {
                messageQueue.put(new DelayedMessage("Immediate message", 0));
                messageQueue.put(new DelayedMessage("Delayed message (2s)", 2000));
                messageQueue.put(new DelayedMessage("Delayed message (1s)", 1000));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();

        // 消息处理器
        new Thread(() -> {
            try {
                while (true) {
                    DelayedMessage message = messageQueue.take();
                    // 实际应用中需要检查是否到达执行时间
                    System.out.println("Processing: " + message.getMessage() +
                            " at " + System.currentTimeMillis());
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

